package saramax

import (
	"encoding/json"
	"errors"
	"gitee.com/xiao_hange/go-admin-pkg/pkg/logger"
	"github.com/IBM/sarama"
)

var KafkaServerErr = errors.New("kafka服务出现故障！")

type Handler[T any] struct {
	l           logger.AppLogger
	consumeName string
	fn          func(msg *sarama.ConsumerMessage, t T) error
}

func NewHandler[T any](l logger.AppLogger, consumeName string, fn func(msg *sarama.ConsumerMessage, t T) error) *Handler[T] {
	return &Handler[T]{
		l:           l,
		fn:          fn,
		consumeName: consumeName,
	}
}

func (h Handler[T]) Setup(session sarama.ConsumerGroupSession) error {
	h.l.Info("KafKa-send-mail-for-csv", logger.String("val", "启动 "+h.consumeName+" 消费者"))
	return nil
}

func (h Handler[T]) Cleanup(session sarama.ConsumerGroupSession) error {
	h.l.Error("致命错误", logger.Error(KafkaServerErr))
	return KafkaServerErr
}

func (h Handler[T]) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	msgs := claim.Messages()
	for msg := range msgs {
		var t T
		err := json.Unmarshal(msg.Value, &t)
		if err != nil {
			h.l.Error("反序列化失败！", logger.Error(err), logger.String("topic", msg.Topic), logger.Int64("offset", msg.Offset), logger.Int32("partition", msg.Partition))
		}
		for i := 0; i < 3; i++ {
			err = h.fn(msg, t)
			if err == nil {
				break
			}
			h.l.Error("处理消息失败！", logger.Error(err), logger.String("topic", msg.Topic), logger.Int64("offset", msg.Offset), logger.Int32("partition", msg.Partition))
		} //简单的重试
		if err != nil {
			h.l.Error("处理消息失败-重试次数上线", logger.Error(err), logger.String("topic", msg.Topic), logger.Int64("offset", msg.Offset), logger.Int32("partition", msg.Partition))
		} else {
			session.MarkMessage(msg, "")
		}
	}
	return nil
}
